package main;


import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.DataInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.LinkedList;
import java.util.List;
import java.util.StringTokenizer;
import java.util.Vector;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.SequenceFileInputFormat;
import ItemMapReduce.*;
import UserMapReduce.*;
import dataStructures.CodeBook;
import dataStructures.FilesAndPaths;
import dataStructures.TwoVectorsFirstIntSecondDouble;
import dataStructures.UsersMapOutputWritable;
import eval.EvalDriver;

import preProcess.PreProcessCombiner;
import preProcess.PreProcessMapper;
import preProcess.PreProcessReducer;
import profiler.ProfilingDriver;


public class Driver {

	/**
	 * @param args
	 */
	//input files
	public static File TRAINING_SET_INPUT_FILE;
	public static File VALIDATION_SET_INPUT_FILE;
	//constant files
	public static File USER_PROFILES_FILE;
	public static File ITEM_PROFILES_FILE;
	public static File USER_PROFILES_VALIDATION_FILE;
	public static File PROFILERS_DIRECTORY;
	//iterative vars

	/*public static File CODEBOOK_USER_FILE;
	public static File CODEBOOK_ITEM_FILE;
	public static File RMSE_FILE;
	public static File CURRENT_ITERATION_DIRECTORY;
	public static File CODEBOOK_USER_DIR;
	public static File CODEBOOK_ITEM_DIR;
	
	public static String CODEBOOK_ITEM_DIR_PATH;*/
	public static int CURRENT_ITERATION = 0;
	//input constants
	public static int K=20; //item codebook cluster size
	public static int L=20; //user codebook cluster size 
	//default values
	public static int T=10;
	public static double EPSILON=0.01;
	//public static int MAXIMUM_ITERATIONS_NUMBER = 10;
	//variables from SV
	public static int Q = -1 ;//users number
	public static int P = -1;//items number
	//prefixes of files
	
	 
 


	public static void main(String[] args) {

		if(args.length != 4 && args.length != 2  && args.length != 6 ){
			System.out.println("args for the system should be in type of: <(ST) , (SV)>");
			System.out.println("args for the system should be in type of: <(ST) , (SV) , (K) , (L)>");
			System.out.println("args for the system should be in type of: <(ST) , (SV) , (K) , (L), (T) , (EPSILON)>");
			System.out.println("Exit program");
			System.exit(-1);
		}
		Configuration conf = new Configuration();
		conf.addResource("/");
		cleanOutputDir(new File(System.getProperty("user.dir").toString()));
		initialization(args);//all args are initialized CodeBookUser was generated
		//System.exit(-1);
		preProcess();
		FilesAndPaths.updateCodeBookItems(CURRENT_ITERATION);
		FilesAndPaths.updateCodeBookUser(CURRENT_ITERATION);
		
		System.out.println("done pre process");

		while(!shouldStop(conf, CURRENT_ITERATION+1)){
			CURRENT_ITERATION++;
			
			FilesAndPaths.createNewPathsForIteration(CURRENT_ITERATION);
			
			userMapRed();
			FilesAndPaths.updateCodeBookUser(CURRENT_ITERATION);
			System.out.println("done user Map Reduce");
			
			itemMapRed();
			FilesAndPaths.updateCodeBookItems(CURRENT_ITERATION);
			System.out.println("done item Map Reduce");
			
			RMSEMapReduce(conf);
			
			
			
			
			
			
		}
		
		
		//TODO : implement method - extract 3 final files
		
		//TODO: uncomment this
		storeFinalFiles(conf,CURRENT_ITERATION); 


	}
	
	
	private static void RMSEMapReduce(Configuration conf){
		conf.setInt("iteration", CURRENT_ITERATION);
		try {
			EvalDriver.runEval(conf);
		} catch (Exception e){
			e.printStackTrace();
			
		}
	}
	
	

	private static void preProcess() {
		JobConf job = new JobConf(new Configuration(), Driver.class);
		job.setInt("K", K);
		job.setInt("L", L);
		try {
			job.set("CodeBookUserPath" , FilesAndPaths.getCodeBookUserFile().getAbsolutePath()  );
			//job.set("CodeBookUserPath" , CODEBOOK_USER_FILE.getAbsolutePath()  );
			Path input = new Path(ITEM_PROFILES_FILE.getAbsolutePath());
		    FileInputFormat.setInputPaths(job, input);
		    job.setInputFormat(SequenceFileInputFormat.class);
		    //SequenceFileInputFormat<LongWritable, ProfileWritable>.
		    
		    Path output = new Path(FilesAndPaths.getCodeBookItemDirPath());
		    //Path output = new Path(CODEBOOK_ITEM_DIR_PATH);
		    FileOutputFormat.setOutputPath(job,output);
		    job.setMapperClass(PreProcessMapper.class);
		    job.setCombinerClass(PreProcessCombiner.class);
		    job.setReducerClass(PreProcessReducer.class);
		    job.setMapOutputKeyClass(IntWritable.class);
		    job.setMapOutputValueClass(UsersMapOutputWritable.class);
		    job.setOutputKeyClass(IntWritable.class);
		    job.setOutputValueClass(TwoVectorsFirstIntSecondDouble.class);
		    System.out.println("pre process configure OK");
		    JobClient.runJob(job);
		} catch (IOException e) {
			System.out.println("Cant read CodeBookUser from file OR job runing problem");
			e.printStackTrace();
		}
	}
	
	private static void userMapRed() {
		JobConf job = new JobConf(new Configuration(), Driver.class);
		job.setInt("K", K);
		job.setInt("L", L);
		try {
			//job.set("CodeBookUserPath" , CODEBOOK_ITEM_FILE.getAbsolutePath()  );
			job.set("CodeBookUserPath" , FilesAndPaths.getCodeBookItemFile().getAbsolutePath());
			Path input = new Path(USER_PROFILES_FILE.getAbsolutePath());
		    FileInputFormat.setInputPaths(job, input);
		    job.setInputFormat(SequenceFileInputFormat.class);

		    Path output = new Path(FilesAndPaths.getCodeBookUserDirPath());
		    FileOutputFormat.setOutputPath(job,output);
		    job.setMapperClass(UserProcessMapper.class);
		    job.setCombinerClass(UserProcessCombiner.class);
		    job.setReducerClass(UserProcessReducer.class);
		    job.setMapOutputKeyClass(IntWritable.class);
		    job.setMapOutputValueClass(UsersMapOutputWritable.class);
		    job.setOutputKeyClass(IntWritable.class);
		    job.setOutputValueClass(TwoVectorsFirstIntSecondDouble.class);

		    System.out.println("User map reduce configure OK");

		    
		    JobClient.runJob(job);
		    
		} catch (IOException e) {
			System.out.println("problem in map reduce user process job runing problem");
			e.printStackTrace();
		}
	}
	
	private static void itemMapRed() {
		JobConf job = new JobConf(new Configuration(), Driver.class);
		job.setInt("K", K);
		job.setInt("L", L);
		try {
			//job.set("CodeBookUserPath" , CODEBOOK_USER_FILE.getAbsolutePath()  );
			job.set("CodeBookUserPath" , FilesAndPaths.getCodeBookUserFile().getAbsolutePath()  );
			Path input = new Path(ITEM_PROFILES_FILE.getAbsolutePath());
		    FileInputFormat.setInputPaths(job, input);
		    job.setInputFormat(SequenceFileInputFormat.class);
		    //SequenceFileInputFormat<LongWritable, ProfileWritable>.
		    Path output = new Path(FilesAndPaths.getCodeBookItemDirPath());
		    FileOutputFormat.setOutputPath(job,output);
		    job.setMapperClass(ItemProcessMapper.class);
		    job.setCombinerClass(ItemProcessCombiner.class);
		    job.setReducerClass(ItemProcessReducer.class);
		    job.setMapOutputKeyClass(IntWritable.class);
		    job.setMapOutputValueClass(UsersMapOutputWritable.class);
		    job.setOutputKeyClass(IntWritable.class);
		    job.setOutputValueClass(TwoVectorsFirstIntSecondDouble.class);
		    System.out.println("User map reduce configure OK");
		    JobClient.runJob(job);
		} catch (IOException e) {
			System.out.println("problem in map reduce user process job runing problem");
			e.printStackTrace();
		}
	}
	
	public static Vector<CodeBook> getDataFromIteration(Configuration conf, int iter, String dir){
		Vector<CodeBook> ans = new Vector<CodeBook>();
		//open relevant file
		FileSystem fileSystem = null;
		try {
			fileSystem = FileSystem.get(conf);
			
			Path rmsePath = new Path("foo" + iter + "/" + dir);
			FileStatus[] status = fileSystem.listStatus(rmsePath);
			for (int i=0;i<status.length;i++){
				if (status[i].getPath().getName().startsWith(FilesAndPaths.MAPREDUCE_FINAL_FILE_NAME_PREFIX)){
					Path path = status[i].getPath();
					FSDataInputStream in = fileSystem.open(path);
					BufferedReader br = new BufferedReader(new InputStreamReader(in));
					String line;
					while ( (line = br.readLine()) != null){
						ans.add(CodeBook.getCBfromString(line));
					}
				}	
			}
		} catch (IOException e){
			e.printStackTrace();
		}		
		return ans;
	}
	
	private static void storeFinalFiles(Configuration conf, int iterations) {
		int bestIteration = findBestIterationByRMSE(conf, iterations);
		Vector<CodeBook> usersData = 
				getDataFromIteration(conf, bestIteration, FilesAndPaths.CODEBOOK_USER_DIR_NAME);
		Vector<CodeBook> itemData = 
				getDataFromIteration(conf, bestIteration, FilesAndPaths.CODEBOOK_ITEM_DIR_NAME);
		
//		int userArr[] = convertVectorToArr(usersData);
		int userArr[] = convertCodeBookVectorToArr(usersData, Q+1);
		int itemArr[] = convertCodeBookVectorToArr(itemData, P+1);
		double codeBookMatrix[][] = 
				convertItemCodeBookVectorToMatrixValues(itemData, K, L );
		//TODO: print arrs to files U.txt, V.txt and B.txt in classpath("/")
		storeCodeBookItems(userArr, "UserClusters.txt", conf);
		storeCodeBookItems(itemArr, "ItemClusters.txt", conf);
		storeCodeBookMatrix(codeBookMatrix, "CodeBook.txt", conf);
		//create U file
		//create V file
		//create B file
		
		
		
	}
	
	public static void storeCodeBookItems(int arr[], String dest, Configuration conf){
		try {
			FileSystem fileSystem = FileSystem.get(conf);
			Path path = new Path("model/" + dest);
		    if (fileSystem.exists(path)) {
		        System.out.println("File " + dest + " already exists");
		        return;
		    }
		    FSDataOutputStream out = fileSystem.create(path);
		    
		    for (int i=1; i<arr.length; i++){
		    	out.writeBytes( i + " , " + arr[i] + "\n");
		    }
		    out.close();
		    fileSystem.close();
		
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	public static void storeCodeBookMatrix(double matrix[][], String dest, Configuration conf){
		try {
			FileSystem fileSystem = FileSystem.get(conf);
			Path path = new Path("model/" + dest);
		    if (fileSystem.exists(path)) {
		        System.out.println("File " + dest + " already exists");
		        return;
		    }
		    FSDataOutputStream out = fileSystem.create(path);
		    
		    for (int i=1; i<matrix.length; i++){
		    	for (int j=1; j<matrix[i].length; j++){
		    		out.writeBytes(Double.toString(matrix[i][j]));
		    		if (j < matrix.length - 1)
		    			out.writeBytes(" , ");
		    	}
		    	out.writeBytes("\n");
		    }
		    out.close();
		    fileSystem.close();
		
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}	
		
	}

	public static int[] convertCodeBookVectorToArr(Vector<CodeBook> codeBookVector, int arrSize) {
		//build ans arr
		int ans[] = new int[arrSize];
		for (int i=0; i<ans.length; i++)
			ans[i] = 0;//(int) Math.random() * Math.min(Driver.K, Driver.L);//-1;for test reasons - we should never see -1 in users!!!
		
		//for each id add the current cluster
		for (CodeBook cb : codeBookVector){
			int clusterId = cb.getClusterId();
			int[] ids = cb.getUserIds();
			for (int i=0; i<ids.length; i++){
				ans[ids[i]] = clusterId;
			}
		}
		
		
		return ans;
	}
	
	/**
	 * by deep copy!!! was not tested, but debugged and should work great!
	 * @param userCodeBookVector
	 * @param itemClusterCount
	 * @param userClusterCount
	 * @return
	 */
	public static double[][] convertUserCodeBookVectorToMatrixValues(Vector<CodeBook> userCodeBookVector,
			int itemClusterCount, int userClusterCount){
		
		double ans[][] = new double[userClusterCount][itemClusterCount];
		//initialize in -1
		for(int i=0; i<userClusterCount; i++){
			for (int j=0; j<itemClusterCount; j++){
				ans[i][j] = -1.0;
			}
		}
		
		for(CodeBook cb: userCodeBookVector){
			double[] codeBookValues = cb.getCodeValues();
			int clusterID = cb.getClusterId();
			//DEEP COPY of code book values
			for(int i=0; i<codeBookValues.length; i++){
				ans[clusterID][i] = codeBookValues[i];
			}
		}
		
		//if there is a cluster with no users or data, change the code book matrix to hold avg value;
		for (int i=0; i<ans.length; i++){
			for (int j=0; j<ans[i].length; j++){
				if(ans[i][j] == -1.0)
					ans[i][j] = 2.5;
			}
		}
		
		
		return ans;
	}
	
	
	/**
	 * by deep copy!!! was not tested, but debugged and should work great!
	 * @param itemCodeBookVector
	 * @param itemClusterCount
	 * @param userClusterCount
	 * @return
	 */
	public static double[][] convertItemCodeBookVectorToMatrixValues(Vector<CodeBook> itemCodeBookVector,
			int itemClusterCount, int userClusterCount){
		
		double ans[][] = new double[userClusterCount][itemClusterCount];
		//initialize in -1
		for(int i=0; i<userClusterCount; i++){
			for (int j=0; j<itemClusterCount; j++){
				ans[i][j] = -1.0;
			}
		}
		
		for(CodeBook cb: itemCodeBookVector){
			double[] codeBookValues = cb.getCodeValues();
			int clusterID = cb.getClusterId();
			//DEEP COPY of code book values
			for(int i=0; i<codeBookValues.length; i++){
				ans[i][clusterID] = codeBookValues[i];
			}
		}
		
		//if there is a cluster with no users or data, change the code book matrix to hold avg value;
		for (int i=0; i<ans.length; i++){
			for (int j=0; j<ans[i].length; j++){
				if(ans[i][j] == -1.0)
					ans[i][j] = 2.5;
			}
		}
		
		
		return ans;
	}
	
	
	
	

	private static int findBestIterationByRMSE(Configuration conf, int iterations) {
		double minRMSE = Double.MAX_VALUE;
		int minIteration = 0; //0 instead of -1 just to be safe
		for (int i=0; i<=iterations; i++){
			double rmse = getRMSE(conf, i);
			if (rmse < minRMSE){
				minRMSE = rmse;
				minIteration = i;
			}
		}
		return minIteration;
	}
	
	private static double getRMSE(Configuration conf, int iteration){
		FileSystem fileSystem = null;
		try {
			fileSystem = FileSystem.get(conf);
			
			Path rmsePath = new Path("foo" + iteration + "/" + FilesAndPaths.RMSE_DIR_NAME);
			FileStatus[] status = fileSystem.listStatus(rmsePath);
			for (int i=0;i<status.length;i++){
				if (status[i].getPath().getName().startsWith(FilesAndPaths.MAPREDUCE_FINAL_FILE_NAME_PREFIX)){
					Path path = status[i].getPath();
					FSDataInputStream in = fileSystem.open(path);
					BufferedReader br = new BufferedReader(new InputStreamReader(in));
					String line;
					double rmse = 0;
					if ( (line = br.readLine()) != null){
						rmse = Double.valueOf(StringUtils.split(line, "\t")[1]);
					}
					br.close();
			        in.close();
			        return rmse;
				}
			}
			
				
		} catch (IOException e){
			e.printStackTrace();
		}
		return Double.MAX_VALUE;
		
	}
	
	

	private static void initialization(String [] args){
		initializeArgs(args); // get K , L and open Files for SV and ST 
		fetchNumberOfUsersAndItems(); // Get P,Q - number of users and items
		
		
		
		buildProfileFilesFromTrainingSet(); // MapReduce Iteration to build 2 profile files and store them in static vars
		buildProfileFilesFromValidationSet();//MapReduce Iteration to build 1 user profile validation and stpre it to static var
		//CURRENT_ITERATION_DIRECTORY = createEmptyDirForIteration(CURRENT_ITERATION);
		FilesAndPaths.addCurrentIterationDir( createEmptyDirForIteration(CURRENT_ITERATION));
		//CODEBOOK_USER_DIR = createEmptyDirForProfiles(CODEBOOK_USER_DIR_NAME , CURRENT_ITERATION_DIRECTORY);
		FilesAndPaths.addCodeBookUserDir(createEmptyDirForProfiles(FilesAndPaths.CODEBOOK_USER_DIR_NAME , FilesAndPaths.getCurrentIterationDir()));
		//CODEBOOK_ITEM_DIR = createEmptyDirForProfiles(CODEBOOK_ITEM_DIR_NAME , CURRENT_ITERATION_DIRECTORY);
		FilesAndPaths.addCodeBookItemDirPath( FilesAndPaths.getCurrentIterationDir() + "/" + FilesAndPaths.CODEBOOK_ITEM_DIR_NAME);
		//CODEBOOK_ITEM_DIR_PATH = CURRENT_ITERATION_DIRECTORY + "/" + CODEBOOK_ITEM_DIR_NAME; 

		//CODEBOOK_USER_FILE = createEmptyCodeBookFile(CODEBOOK_USER_DIR, MAPREDUCE_FINAL_FILE_NAME_PREFIX + "00000");// createEmptyUserClustersFile(dir);
		FilesAndPaths.addCodeBookUserFile(createEmptyCodeBookFile(FilesAndPaths.getCodeBookUserDir(), FilesAndPaths.MAPREDUCE_FINAL_FILE_NAME_PREFIX + "00000"));// createEmptyUserClustersFile(dir);
		//CODEBOOK_ITEM_FILE = createEmptyCodeBookFile(dirOfCodebookItem, NAME_OF_MAPREDUCE_FINAL_FILE); //createEmptyItemClustersFile(dir);
		//generateInitialDummyCodeBookUserFile(CODEBOOK_USER_FILE, Q, L);
		generateInitialDummyCodeBookUserFile(FilesAndPaths.getCodeBookUserFile(), Q, L);

	}

	private static void buildProfileFilesFromTrainingSet(){
		//String parentDir = TRAINING_SET_INPUT_FILE.getParent();
		if(PROFILERS_DIRECTORY == null)
			PROFILERS_DIRECTORY = new File(System.getProperty("user.dir").toString() + "/" + FilesAndPaths.PROFILERS_DIR_NAME) ;
		String [] args = new String[3];
		args [0] = TRAINING_SET_INPUT_FILE.getAbsolutePath();
		String userProfileDir =  FilesAndPaths.PROFILERS_DIR_NAME + "//" + "userProfile";
		String itemProfileDir =  FilesAndPaths.PROFILERS_DIR_NAME + "//" + "itemProfile";
		args [1] = userProfileDir;
		args [2] = itemProfileDir;
		try {
			//System.out.println("here1");
			ProfilingDriver.getInstance().profile(args);
			//System.out.println("here2");
			USER_PROFILES_FILE = updateStaticVarsOfFiles(userProfileDir);
			ITEM_PROFILES_FILE = updateStaticVarsOfFiles(itemProfileDir);
			//System.out.println("here3");

		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	private static void buildProfileFilesFromValidationSet(){
		//String parentDir = TRAINING_SET_INPUT_FILE.getParent();
		//String parentDir = VALIDATION_SET_INPUT_FILE.getParent();
		
		if(PROFILERS_DIRECTORY == null)
			PROFILERS_DIRECTORY = new File(System.getProperty("user.dir").toString() + "/" + FilesAndPaths.PROFILERS_DIR_NAME) ;
		String [] args = new String[3];
		args [0] = VALIDATION_SET_INPUT_FILE.getAbsolutePath();
		String userProfileValidationDir =  FilesAndPaths.PROFILERS_DIR_NAME + "//" + "userProfileValidation";
		String itemProfileValidationDir =  FilesAndPaths.PROFILERS_DIR_NAME + "//" + "itemProfileValidation";
		args [1] = userProfileValidationDir;
		args [2] = itemProfileValidationDir;
		try {
			//System.out.println("here1");
			ProfilingDriver.getInstance().profile(args);
			//System.out.println("here2");
			USER_PROFILES_VALIDATION_FILE = updateStaticVarsOfFiles(userProfileValidationDir);
			//ITEM_PROFILES_FILE = updateStaticVarsOfFiles(itemProfileDir);
			//System.out.println("here3");

		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	private static File updateStaticVarsOfFiles(String dir) {
		File directory = new File(dir);
		String name;
		File[] listFiles = directory.listFiles();
		for(int i = 0 ; i < listFiles.length ; i++){
			name = listFiles[i].getName();
			if(name.startsWith(FilesAndPaths.MAPREDUCE_FINAL_FILE_NAME_PREFIX) && !name.startsWith(".")){
				return listFiles[i];
			}		
		}

		throw new RuntimeException("DIDNT FOUND MAPREDUCE FILE");
	}

	private static void generateInitialDummyCodeBookUserFile(File f, int number1, int rangeInterval ) {
		int r;
		List <Integer> IDs;
		List <Integer> codeBook;
		int [] clusterOfID = new int [number1+1];   
		try {
			BufferedWriter output = new BufferedWriter(new FileWriter(f));
			for(int i = 1 ; i<=number1 ; i++){
				r = (int)(Math.random() * rangeInterval);
				clusterOfID[i] = r;
			}
			//Each ID is assigned to cluster 
			for(int j = 0 ; j <= rangeInterval-1 ; j++){
				IDs = new LinkedList<Integer>();
				codeBook = new LinkedList<Integer>();
				for(int n = 1 ; n <= number1 ; n++){
					if(clusterOfID[n] == j){
						IDs.add(n);
					}
				}
				
				output.write(j + "\t" + IDs.toString() + "\t" + codeBook.toString()+"\n");
			}

			output.close();
		} catch ( IOException e ) {
			e.printStackTrace();
		}
	}

	private static void initializeArgs(String[] args) {
		TRAINING_SET_INPUT_FILE = new File(args[0]);
		VALIDATION_SET_INPUT_FILE = new File(args[1]);
		if(args.length == 4 || args.length == 6){
		K=Integer.valueOf(args[2]); 
		L=Integer.valueOf(args[3]);
		}
		
		if(args.length == 6){
			T=Integer.valueOf(args[4]); 
			EPSILON=Double.valueOf(args[5]);
			}
		
		
		
	}

	private static void fetchNumberOfUsersAndItems() {
		StringTokenizer st;
		int userID , itemID;
		try{

			FileInputStream fstream = new FileInputStream(TRAINING_SET_INPUT_FILE);
			DataInputStream in = new DataInputStream(fstream);
			BufferedReader br = new BufferedReader(new InputStreamReader(in));
			String strLine;
			//Read File Line By Line
			while ((strLine = br.readLine()) != null)   {
				st = new StringTokenizer(strLine, " \t\n");
				userID = Integer.valueOf(st.nextToken());
				if(userID > Q)
					Q = userID;
				itemID = Integer.valueOf(st.nextToken());
				if(itemID > P)
					P = itemID;

			}
			//Close the input stream
			in.close();
		}catch (Exception e){//Catch exception if any
			System.err.println("Error: " + e.getMessage());
		}
	}

	private static boolean shouldStop(Configuration conf, int currentIter) {
		if(currentIter == 0 || currentIter == 1)
			return false; // dont stop
		if(currentIter > T)
			return true; //stop
		if(getRmseDiff(conf, currentIter) < EPSILON)
			return true; //stop
		
		return false; 

	}

	private static double getRmseDiff(Configuration conf, int currentIter) {
		int prevIter = currentIter - 1;
		double prevIterRMSE = getRMSE(conf, prevIter);
		double currIterRMSE = getRMSE(conf, currentIter);
		return Math.abs(prevIterRMSE - currIterRMSE);	
	}

	private static File createEmptyCodeBookFile(File dir , String prefix ) {
		String userClustersFilePath = dir.getAbsolutePath() + "/" +  prefix;
		return new File (userClustersFilePath);
	}

	private static File createEmptyDirForIteration(int iteration) {
		String directoryName = "foo" + iteration;
		File dir = new File(directoryName);
		
		if (dir.exists()){
			try {
				FileUtils.deleteDirectory(dir);
			} catch (IOException e) {
				System.out.println("Problem with deletion");
				System.exit(-1);
			}
		}
			
		System.out.println("creating directory: " + directoryName);
		dir.mkdir();
		System.out.println("the dir " + directoryName + " was created");
		return dir;

	}
	
	private static File createEmptyDirForProfiles(String dir , File itertionDir) {
		String directoryName = itertionDir + "/" + dir;
		File newDir = new File(directoryName);
		
		if (newDir.exists()){
			try {
				FileUtils.deleteDirectory(newDir);
			} catch (IOException e) {
				System.out.println("Problem with deletion");
				System.exit(-1);
			}
		}
			System.out.println("creating directory: " + directoryName);
			newDir.mkdir();			
			System.out.println("the dir " + directoryName + " was created");

		return newDir;

	}
	
	private static void cleanOutputDir(File cwd){
		File[] listFiles = cwd.listFiles();
		for(File dir : listFiles){
		if (dir.isDirectory()  && ( dir.getName().startsWith("foo") || dir.getName().startsWith("Profilers") )){
			try {
				FileUtils.deleteDirectory(dir);
			} catch (IOException e) {
				System.out.println("Problem with deletion");
				System.exit(-1);
			}
		}
	}
	}
	
	
	

}
